package com.ecommerce.recommendation.util;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.springframework.stereotype.Component;
import org.springframework.web.multipart.MultipartFile;

import java.io.IOException;
import java.io.InputStream;
import java.util.*;

@Component
public class HdfsUtil {

    private FileSystem fileSystem;

    public HdfsUtil() throws IOException {
        System.setProperty("hadoop.home.dir", "E:\\hadoop-3.2.0");
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://192.168.74.100:9000");
        this.fileSystem = FileSystem.get(conf);
    }

    public String uploadFile(MultipartFile file) throws IOException {
        if (file.isEmpty()) {
            throw new IOException("上传的文件为空");
        }

        String originalFilename = file.getOriginalFilename();
        Path hdfsPath = new Path("/user/hadoop/ecommerce/input/" + originalFilename);

        // 确保目标目录存在
        Path parentDir = hdfsPath.getParent();
        if (!fileSystem.exists(parentDir)) {
            fileSystem.mkdirs(parentDir);
        }

        try (InputStream inputStream = file.getInputStream();
             FSDataOutputStream outputStream = fileSystem.create(hdfsPath)) {

            byte[] buffer = new byte[4096];
            int bytesRead;

            while ((bytesRead = inputStream.read(buffer)) != -1) {
                outputStream.write(buffer, 0, bytesRead);
            }
        }

        return hdfsPath.toString();
    }

    public Map<String, Object> readAnalysisResult(String directoryPath) throws IOException {
        Path path = new Path(directoryPath);
        Map<String, Object> result = new HashMap<>();

        if (fileSystem.exists(path) && fileSystem.isDirectory(path)) {
            FileStatus[] fileStatuses = fileSystem.listStatus(path);
            List<Map<String, Object>> data = new ArrayList<>();

            for (FileStatus status : fileStatuses) {
                if (!status.getPath().getName().startsWith("part-")) {
                    continue;
                }

                try (FSDataInputStream inputStream = fileSystem.open(status.getPath())) {
                    Scanner scanner = new Scanner(inputStream);
                    while (scanner.hasNextLine()) {
                        String line = scanner.nextLine();
                        String[] parts = line.split("\t");
                        if (parts.length >= 2) {
                            Map<String, Object> record = new HashMap<>();
                            record.put("key", parts[0]);
                            record.put("value", parts[1]);
                            data.add(record);
                        }
                    }
                }
            }

            result.put("data", data);
            result.put("status", "success");
        } else {
            result.put("status", "error");
            result.put("message", "结果目录不存在");
        }

        return result;
    }

    public void close() {
        if (fileSystem != null) {
            try {
                fileSystem.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}